package com.ethink.framework.redis.mq;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.thread.ThreadUtil;
import com.ethink.framework.redis.properties.RedisMqProperties;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author name(Tel : mobile)
 * @Date 2020-08-13 09:46
 **/
public class RedisMqListenerRegister {

    private RedisMqProperties redisMqProperties;
    private RedisMqPublisher publisher;
    private RedisTemplate redisTemplate;
    private Map<String, ConsumerThread> topicMap = new HashMap<>();

    public void register(String topic, boolean retryable, RedisMqMessageHandler mqMessageHandler,
                         int corePoolSize, int maximumPoolSize, int queueSize) {
        ExecutorService executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 1000L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(queueSize), ThreadFactoryBuilder.create().setNamePrefix("rmq-" + topic + "-").build());
        ConsumerThread consumerThread = new ConsumerThread();
        consumerThread.setTopic(topic);
        consumerThread.setRetryable(retryable);
        consumerThread.setMqMessageHandler(mqMessageHandler);

        consumerThread.setExecutorService(executorService);
        consumerThread.setRedisTemplate(redisTemplate);
        consumerThread.setPublisher(publisher);
        consumerThread.setEmptySleepTime(redisMqProperties.getEmptySleepTime());
        consumerThread.setErrorSleepTime(redisMqProperties.getErrorSleepTime());
        topicMap.put(topic, consumerThread);
        consumerThread.start();
    }

    public void setRedisMqProperties(RedisMqProperties redisMqProperties) {
        this.redisMqProperties = redisMqProperties;
    }

    public void setPublisher(RedisMqPublisher publisher) {
        this.publisher = publisher;
    }

    public void setRedisTemplate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Data
    @Slf4j
    public static class ConsumerThread extends Thread {
        private String topic;
        private boolean retryable;
        private RedisMqMessageHandler mqMessageHandler;

        private ExecutorService executorService;
        private RedisTemplate redisTemplate;
        private RedisMqPublisher publisher;

        private long emptySleepTime = 1000L;
        private long errorSleepTime = 1000L;

        private long size = 0;

        @Override
        public void run() {
            while (true) {
                try {
                    if (size <= 0) {
                        ThreadUtil.sleep(emptySleepTime);
                        Long remoteSize = redisTemplate.opsForList().size(MqRedisRedisConstant.TOPIC_PREFIX + topic);
                        if (remoteSize == null) {
                            size = 0;
                        } else {
                            size = remoteSize;
                        }
                    } else {
                        Object message = redisTemplate.opsForList().leftPop(MqRedisRedisConstant.TOPIC_PREFIX + topic);
                        if (message == null) {
                            size = 0;
                        } else {
                            size--;
                            try {
                                executorService.execute(() -> {
                                    try {
                                        mqMessageHandler.handler(message);
                                    } catch (Exception e) {
                                        log.error("rmq handler error", e);
                                        if (retryable) {
                                            try {
                                                publisher.publish(topic, message);
                                            } catch (Exception e2) {
                                                log.error("rmq message publish error", e);
                                            }
                                        }
                                    }
                                });
                            } catch (Exception e) {
                                log.error("rmq submit job error", e);
                                try {
                                    publisher.publish(topic, message);
                                } catch (Exception e2) {
                                    log.error("rmq message publish error", e);
                                }
                            }
                        }
                    }
                } catch (Error e) {
                    log.error("rmq error", e);
                    ThreadUtil.sleep(errorSleepTime);
                }
            }
        }
    }
}
